package com.hzlj.position.job.listener;

import cn.hutool.core.util.ObjectUtil;
import com.fz.common.base.core.JsonUtils;
import com.fz.common.kafka.annotation.KafkaListenerRetry;
import com.hzlj.position.config.common.dto.operatorActivation.OperatorActivationDTO;
import com.hzlj.position.config.common.dto.operatorActivation.OperatorActivationQueryParamDTO;
import com.hzlj.position.config.common.dto.operatorActivation.OperatorActivationQueryResultDTO;
import com.hzlj.position.config.common.enums.OperatorActivationStatus;
import com.hzlj.position.config.service.OperatorActivationService;
import com.hzlj.sqjz.jz.common.dto.common.JzCommonQueryResultDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

import static com.hzlj.sqjz.config.common.constant.Topic.JZ_JJ;


/**
 * @author lifh
 * @date 2023/3/17 11:29
 **/
@Slf4j
@Component
public class JzryJjListener {
    @Resource
    private OperatorActivationService operatorActivationService;

    @KafkaListenerRetry(enable = false)
    @KafkaListener(topics = {JZ_JJ})
    public void jzryJj(ConsumerRecord<String, JzCommonQueryResultDTO> record) {
        log.info("【kafka】收到解矫消息:{}", record.value());
        JzCommonQueryResultDTO jbxx = record.value();
        //取消白名单
        try {
            List<OperatorActivationQueryResultDTO> actives = operatorActivationService.listOperatorActivation(new OperatorActivationQueryParamDTO()
                    .setJzId(jbxx.getJzId())
                    .setActivation(OperatorActivationStatus.ACTIVATION.getCode())
            );
            if (ObjectUtil.isEmpty(actives)) {
                log.info("收到解矫消息，没有白名单开通信息，不需要取消白名单:jbxx={}", JsonUtils.obj2json(jbxx));
                return;
            }

            for (OperatorActivationQueryResultDTO active : actives) {
                log.info("收到解矫消息，取消白名单:sjh={}", active.getSjh());
                OperatorActivationDTO update = new OperatorActivationDTO()
                        .setId(active.getId())
                        .setJgCode(active.getJgCode())
                        .setChannel(active.getChannel())
                        .setSjh(active.getSjh())
                        .setActivation(OperatorActivationStatus.DEACTIVATE.getCode());
                operatorActivationService.activation(update);
                log.info("收到解矫消息，取消白名单成功:sjh={},active={},jbxx={}", active.getSjh(), JsonUtils.obj2json(active), JsonUtils.obj2json(jbxx));
            }
        } catch (Exception e) {
            log.info("【kafka】解矫消息同步失败(取消白名单):{}", jbxx, e);
        }
    }
}
